package eiisan.forkjointest.countedcompleter;

import java.util.concurrent.CountedCompleter;

public class MapReducer<E> extends CountedCompleter<E> {
    final E[] array;
    final MyMapper<E> mapper;
    final MyReducer<E> reducer;
    final int lo, hi;
    MapReducer<E> sibling;
    public E result;

    public MapReducer(CountedCompleter<?> p, E[] array, MyMapper<E> mapper,
                      MyReducer<E> reducer, int lo, int hi) {
        super(p);
        this.array = array;
        this.mapper = mapper;
        this.reducer = reducer;
        this.lo = lo;
        this.hi = hi;
    }

    public void compute() {
        if (hi - lo >= 2) {
            int mid = (lo + hi) >>> 1;
            MapReducer<E> left = new MapReducer(this, array, mapper, reducer, lo, mid);
            MapReducer<E> right = new MapReducer(this, array, mapper, reducer, mid, hi);
            left.sibling = right;
            right.sibling = left;
            setPendingCount(1); // only right is pending
            right.fork();
            left.compute();     // directly execute left
        } else {
            if (hi > lo) {
                result = mapper.apply(array[lo]);

            }
            //它会依次调用onCompletion.并且是自己调自己或completer调子,
            //且只有左右两个子后完成的能调成功(父任务的挂起数达到0).
            tryComplete();
        }
    }

    public void onCompletion(CountedCompleter<?> caller) {
        //忽略自己调自己.
        if (caller != this) {
            //参数是子任务.
            MapReducer<E> child = (MapReducer<E>) caller;
            MapReducer<E> sib = child.sibling;
            //设置父的result.
            if (sib == null || sib.result == null)
                result = child.result;
            else
                result = reducer.apply(child.result, sib.result);
        }
    }

    public E getRawResult() {
        return result;
    }

    public static <E> E mapReduce(E[] array, MyMapper<E> mapper, MyReducer<E> reducer) {
        return new MapReducer<E>(null, array, mapper, reducer,
                0, array.length).invoke();
    }

}
